/**
 * 
 */
package skewreduce.framework.planner;

import java.util.ArrayList;
import java.util.List;

import skewreduce.framework.Schedule;
import skewreduce.framework.logical.LogicalOp;
import skewreduce.framework.physical.PhysicalOp;
import skewreduce.lib.ITuple;
import skewreduce.lib.Partition;


public class PartitionNode implements Comparable<PartitionNode> {
	final Partition partition;
	double workCost;
	double mergeCost = Double.NaN;
	
	double workSecs;
	double mergeSecs;
	
	double endAt;
	List<ITuple> sample;
	// FIXME: partition strategy
	
	final PartitionNode parent;
	PartitionNode left;
	PartitionNode right;
	int slacks;
	
	enum Status {
		NOT_READY,
		READY,
		COMPLETE
	}
	Status status;
	
	public PartitionNode(Partition p,List<ITuple> sample) {
		this(p,sample,true);
	}
	
	public PartitionNode(Partition p,List<ITuple> sample,boolean alloc) {
		this.parent = null;
		partition = p;
		this.sample = ( alloc ) ? new ArrayList<ITuple>(sample) : sample;
	}
	
	public PartitionNode(PartitionNode parent,Partition p,List<ITuple> sample) {
		this(parent,p,sample,Double.NaN, Double.NaN);
	}
	
	public PartitionNode(PartitionNode parent,Partition p,List<ITuple> sample,double cost, double r) {
		this.parent = parent;
		partition = p;
		this.sample = new ArrayList<ITuple>(sample);
		this.workCost = cost;
		this.slacks = parent.slacks; // copy parents' slack count
		workSecs = ( Double.isNaN(cost) ) ? Double.NaN : cost * r;
	}
	
	public void schedule(double at,double delay) {
		//endAt = at + delay + (( isLeaf() ) ? workCost : mergeCost);
		endAt = at + delay + (( isLeaf() ) ? workSecs : mergeSecs);
	}
	public void complete() {
		status = Status.COMPLETE;
	}
	public boolean isReady() {
		return isLeaf() || (left.status == Status.COMPLETE && right.status == Status.COMPLETE);
	}
	public void resetStatus() {
		status = Status.NOT_READY;
	}
	
	public Partition getPartition() { return partition; }
	public double getWorkCost() { return workCost; }
	public double getMergeCost() { return mergeCost; }
	public double getSecs() {
		return ( isLeaf() ) ? workSecs : mergeSecs;
	}
	public double getWorkSecs() { return workSecs; }
	public double getMergeSecs() { return mergeSecs; }
	public double getCost() {
		return ( isLeaf() ) ? workCost : mergeCost;
	}
	public double getTotalMinCost(double overhead) {
		return Math.max( left.getCost(), right.getCost() ) + mergeCost + overhead;
	}
	
	public double getTotalMinSecs(double delay) {
		return Math.max( left.getSecs(), right.getSecs() ) + mergeSecs + delay;
	}
	
	public List<ITuple> getSample() { return sample; }
	public double getEndAt() { return endAt; }
	public long getExpectedItems(double sampleRate) {
		return (long) (sample.size()/sampleRate);
	}
	public PhysicalOp.ID getOpID() {
		return new PhysicalOp.ID(partition);
	}

	public PartitionNode getParent() { return parent; }
	public PartitionNode getLeft() { return left; }
	public PartitionNode getRight() { return right; }
	public Schedule getSchedule() {
		return new Schedule( new PhysicalOp.ID(partition), getSecs(), getCost() );
	}
	
	public int getSlack() { return slacks; }
	public void setSlack(int n) { slacks = n; }
	
	public boolean isRoot() { return parent == null; }
	public boolean isLeaf() { return left == null && right == null; }
	public boolean canPartition() { return isLeaf() && sample.size() > 2*partition.getArity(); }
	
	public void setWorkCost(double cost, double r) {
		this.workCost = cost;
		workSecs = cost * r;
	}
	public void setMergeCost(double cost, double r) {
		this.mergeCost = cost;
		mergeSecs = cost * r;
	}
	public void setTimes(EaggCostModel model) {
		workSecs = workCost * model.getWork2Sec();
		mergeSecs = mergeCost * model.getMerge2Sec();
	}
	
	public boolean hasWorkCost() { return ! Double.isNaN(workCost); }
	
	public void setLeft(PartitionNode l) { left = l; }
	public void setRight(PartitionNode r) { right = r; }
	public void setChildren(PartitionNode... children) {
		left = children[0];
		right = children[1];
	}
	
	public void rollback() {
		left = null;
		right = null;
		partition.setLeft(null);
		partition.setRight(null);
		mergeCost = Double.NaN;
	}
	
	public void trim() {
		if ( Double.isNaN(mergeCost) || isLeaf() ) {
			throw new IllegalStateException("Can not drop sample from node "+partition.getPartitionID());
		}
		this.sample = null;
	}
	
	@Override
	public int hashCode() { return partition.hashCode(); }
	
	@Override
	public boolean equals(Object o) {
		if ( this == o ) {
			return true;
		} else if ( o instanceof PartitionNode ) {
			return partition.equals( ((PartitionNode)o).partition);
		} else if ( o instanceof Partition ) {
			return partition.equals(o);
		}
		return false;
	}
	
	/**
	 * Ordering
	 * First, descending order in COST
	 * Second, descending order in LEVEL
	 * Thus, expensive operator goes first then deeper level goes first.
	 */
	@Override
	public int compareTo(PartitionNode o) {
		boolean thisLeaf = isLeaf();
		boolean otherLeaf = o.isLeaf();
		if ( thisLeaf == otherLeaf ) {
			return ( getCost() == o.getCost() ) ? partition.compareTo(o.partition) : (int)(Math.signum(o.getCost() - getCost()));
		} else if ( thisLeaf ) {
			return -1;
		}
		return 1;
	}
}